package com.sscw.tcp.mqtt;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.mqtt.server.callback.MessageListener;
import com.alibaba.mqtt.server.model.MessageProperties;
import com.guaike.core.redis.annotation.TryLock;
import com.guaike.core.redis.enums.LockModel;
import com.sscw.tcp.dao.GConfDeviceDAO;
import com.sscw.tcp.dao.GConfRoomDeviceDAO;
import com.sscw.tcp.dao.GRecordGameDAO;
import com.sscw.tcp.dao.GRecordMqttMessageDAO;
import com.sscw.tcp.constant.MQTTTopicConstant;
import com.sscw.tcp.constant.TcpCommandCodeConstant;
import com.sscw.tcp.enums.MessageTypeEnum;
import com.sscw.tcp.enums.RoomTypeEnum;
import com.sscw.tcp.entity.GConfDevicePO;
import com.sscw.tcp.entity.GConfRoomDevicePO;
import com.sscw.tcp.entity.GRecordGamePO;
import com.sscw.tcp.entity.GRecordMqttMessagePO;
import com.sscw.tcp.service.MQTTService;
import com.sscw.tcp.utils.TcpUtils;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;

@Slf4j
@Component
@AllArgsConstructor
public class AppListener implements MessageListener {
    private static final Set<String> commandCodeSet = new HashSet<>(Arrays.asList(MQTTTopicConstant.MACHINE_UP, MQTTTopicConstant.MACHINE_DOWN, MQTTTopicConstant.MACHINE_LEFT, MQTTTopicConstant.MACHINE_RIGHT
            , MQTTTopicConstant.MACHINE_PUSH_COINS, MQTTTopicConstant.MACHINE_ADD_FIRE, MQTTTopicConstant.MACHINE_OPEN_FIRE));
    private final MQTTService mqttService;
    private final GRecordMqttMessageDAO gRecordMqttMessageDAO;
    private final GRecordGameDAO gRecordGameDAO;
    private final GConfDeviceDAO gConfDeviceDAO;
    private final GConfRoomDeviceDAO gConfRoomDeviceDAO;
    private final MQTTProducer messageProducer;

    @SneakyThrows
    @Override
    @TryLock(lockKey = "#msgId", lockModel = LockModel.REENTRANT)
    public void process(String msgId, MessageProperties messageProperties, byte[] payload) {
        try {
            String msg = new String(payload, StandardCharsets.UTF_8);
            log.info("Receive message : msgId = {}, messageProperties = {}, payload = {}", msgId, JSONObject.toJSONString(messageProperties), msg);
            if (!commandCodeSet.contains(messageProperties.getSecondTopic())) {
                return;
            }
            if (MQTTTopicConstant.MACHINE_PUSH_COINS.equals(messageProperties.getSecondTopic())) {
                GRecordMqttMessagePO messagePO = gRecordMqttMessageDAO.findOne(GRecordMqttMessagePO::getMessageId, msgId);
                if (Objects.nonNull(messagePO)) {
                    return;
                }
            }
            GConfDevicePO devicePO = gConfDeviceDAO.findOne(GConfDevicePO::getSerialNumber, msg);
            if (Objects.isNull(devicePO)) {
                return;
            }
            if (!(devicePO.getDeviceType().equals(RoomTypeEnum.arcade_machine.getCode()) || devicePO.getDeviceType().equals(RoomTypeEnum.recreation.getCode()))) {
                return;
            }
            GConfRoomDevicePO roomDevicePO = gConfRoomDeviceDAO.findOne(GConfRoomDevicePO::getDeviceId, devicePO.getId());
            if (Objects.isNull(roomDevicePO)) {
                return;
            }
            GRecordGamePO gamePO = gRecordGameDAO.findRoomPlaying(roomDevicePO.getRoomId(), roomDevicePO.getPositionNo());
            if (Objects.isNull(gamePO)) {
                return;
            }
            String command;
            String stopCommand = null;
            GRecordMqttMessagePO messagePO = null;
            if (MQTTTopicConstant.MACHINE_PUSH_COINS.equals(messageProperties.getSecondTopic())) {
                messagePO = new GRecordMqttMessagePO();
                messagePO.setMessageId(msgId);
                messagePO.setParentTopic(messageProperties.getFirstTopic());
                messagePO.setChildTopic(messageProperties.getSecondTopic());
                messagePO.setMessage(msg);
                messagePO.setStatus(Boolean.FALSE);
                messagePO.setCommandCode(messageProperties.getSecondTopic());
                messagePO.setSerialNumber(msg);
                messagePO.setMessageType(MessageTypeEnum.game.getCode());
                messagePO.setGameId(gamePO.getId());
                gRecordMqttMessageDAO.save(messagePO);
                command = TcpCommandCodeConstant.PUSH_COINS;
            } else if (MQTTTopicConstant.MACHINE_UP.equals(messageProperties.getSecondTopic())) {
                if (!devicePO.getDeviceType().equals(RoomTypeEnum.arcade_machine.getCode())) {
                    return;
                }
                command = TcpCommandCodeConstant.UP;
                stopCommand = TcpCommandCodeConstant.STOP;
            } else if (MQTTTopicConstant.MACHINE_UP_LONG_PRESS.equals(messageProperties.getSecondTopic())) {
                if (!devicePO.getDeviceType().equals(RoomTypeEnum.arcade_machine.getCode())) {
                    return;
                }
                command = TcpCommandCodeConstant.UP;
            } else if (MQTTTopicConstant.MACHINE_DOWN.equals(messageProperties.getSecondTopic())) {
                if (!devicePO.getDeviceType().equals(RoomTypeEnum.arcade_machine.getCode())) {
                    return;
                }
                command = TcpCommandCodeConstant.DOWN;
                stopCommand = TcpCommandCodeConstant.STOP;
            } else if (MQTTTopicConstant.MACHINE_DOWN_LONG_PRESS.equals(messageProperties.getSecondTopic())) {
                if (!devicePO.getDeviceType().equals(RoomTypeEnum.arcade_machine.getCode())) {
                    return;
                }
                command = TcpCommandCodeConstant.DOWN;
            } else if (MQTTTopicConstant.MACHINE_LEFT.equals(messageProperties.getSecondTopic())) {
                if (!devicePO.getDeviceType().equals(RoomTypeEnum.arcade_machine.getCode())) {
                    return;
                }
                command = TcpCommandCodeConstant.LEFT;
                stopCommand = TcpCommandCodeConstant.STOP;
            } else if (MQTTTopicConstant.MACHINE_RIGHT.equals(messageProperties.getSecondTopic())) {
                if (!devicePO.getDeviceType().equals(RoomTypeEnum.arcade_machine.getCode())) {
                    return;
                }
                command = TcpCommandCodeConstant.RIGHT;
                stopCommand = TcpCommandCodeConstant.STOP;
            } else if (MQTTTopicConstant.MACHINE_ADD_FIRE.equals(messageProperties.getSecondTopic())) {
                if (devicePO.getDeviceType().equals(RoomTypeEnum.arcade_machine.getCode())) {
                    command = TcpCommandCodeConstant.MACHINE_ADD_FIRE;
                } else {
                    command = TcpCommandCodeConstant.RECREATION_ADD_FIRE;
                }
            } else if (MQTTTopicConstant.MACHINE_OPEN_FIRE.equals(messageProperties.getSecondTopic())) {
                if (devicePO.getDeviceType().equals(RoomTypeEnum.arcade_machine.getCode())) {
                    command = TcpCommandCodeConstant.MACHINE_OPEN_FIRE;
                } else {
                    command = TcpCommandCodeConstant.RECREATION_OPEN_FIRE;
                }
            } else {
                return;
            }
            Socket socket = TcpUtils.findSocket(msg);
            TcpUtils.sendMessage(socket, command);
            if (StringUtils.isNotBlank(stopCommand)) {
                TcpUtils.sendMessage(socket, stopCommand);
            }
            if (MQTTTopicConstant.MACHINE_PUSH_COINS.equals(messageProperties.getSecondTopic())) {
                mqttService.dealPushCoins(messagePO);
                messageProducer.sendMessage(MQTTTopicConstant.MACHINE_PUSH_COINS_RESULT, gamePO.getUserId().toString());
            }
        } catch (Exception e) {
            log.error("AppListener error={}", e);
            throw new Exception(e);
        }
    }
}
